package com.microcity.rxjava2exe;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

/**
 * RxJava2 基本使用
 *
 * 1.上游可以发送无限个onNext, 下游也可以接收无限个onNext.
 * 2.当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
 * 3.当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
 * 上游可以不发送onComplete或onError.
 * 4.最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
 *
 * 注: 关于onComplete和onError唯一并且互斥这一点, 是需要自行在代码中进行控制, 如果你的代码逻辑中违背了这个规则,
 * 并不一定会导致程序崩溃. 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了,
 * 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.
 */

public class MainActivity extends AppCompatActivity {

    private String TAG = "MY" + this.getClass().getSimpleName();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        //非链式
//        initRx1();
        //链式
//        initRx2();
        //链式Consumer
        initRx3();
    }

    private void initRx3() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "Consumer---->" + integer);//代表上游执行了 (3)
            }
        });
    }

    /**
     * ObservableEmitter： Emitter是发射器的意思  用来分发事件
     *  Disposable：用完即丢弃 切断(Subscribe)两根管子间的连接
     * 注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件.
     */
    private void initRx1() {
        //(1)上游
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });
        //(2)下游
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");//代表上游执行了 (3)
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "onNext---->" + value);//上游执行onNext 发射事件
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");//上游执行onComplete 发射事件
            }
        };
        //(3)上游Subscribe下游
        observable.subscribe(observer);
    }

    private void initRx2() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");//代表上游执行了 (3)
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "onNext---->" + value);//上游执行onNext 发射事件
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");//上游执行onComplete 发射事件
            }
        });
    }


}
